package org.geekbang.work9;

import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import java.util.EventObject;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTextMessage;

/**
 * activemq distributed event publisher.
 * @author leo
 * @date 2021/8/29
 * @since 1.0.0
 */
public class ActiveMQDistributeEventPublisher implements DistributedEventPublisher {

    /**
     * top connection
     */
    private final TopicConnection topicConnection;

    /**
     * topic name
     */
    private final String topicName;

    /**
     * 线程池
     */
    private final ExecutorService executorService;

    /**
     * 消息发布
     */
    private final Publisher<EventObject> publisher;

    public ActiveMQDistributeEventPublisher(String brokerURL, String topicName,Publisher<EventObject> publisher) {
        this.publisher = publisher;
        TopicConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
        //创建连接并启动
        try {
            this.topicConnection = connectionFactory.createTopicConnection();
            this.topicConnection.start();
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }
        this.topicName = topicName;
        //监听DistributeEventObject
        addListener((event) -> {
            //远程发送
            if (event instanceof DistributedEventObject){
                publish((DistributedEventObject) event);
            }
        });

        executorService = Executors.newSingleThreadExecutor();
        //接收 Activemq 发送的消息本地处理
        executorService.submit(this::receiveActiveMQMessage);
    }

    /**
     * 功能: 接收Activemq 消息.
     * @author leo
     * @date 2021/8/29
    */
    private void receiveActiveMQMessage(){
        try {
            TopicSession session = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(this.topicName);
            TopicSubscriber subscriber = session.createSubscriber(topic);
            subscriber.setMessageListener((message) ->{
                try {
                    //发送本地消息
                    publish(new EventObject(((TextMessage)message).getText()));
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    /**
     * 功能: 发布分布式消息.
     * @author leo
     * @date 2021/8/29
     * @param eventObject {@link DistributedEventObject}
     *
    */
    private void publish(DistributedEventObject eventObject) {
        try {
            TopicSession session = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(this.topicName);
            TopicPublisher topicPublisher = session.createPublisher(topic);
            ActiveMQTextMessage message = new ActiveMQTextMessage();
            message.setText(String.valueOf(eventObject.getSource()));
            topicPublisher.publish(message);
            System.out.printf("distributed event publish :%s \n",eventObject);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布本地消息.
     * @param eventObject eventObject
     */
    private void publish(EventObject eventObject) {
        this.publisher.publish(eventObject);
        System.out.printf("local event publish :%s \n",eventObject);
    }

    @Override
    public void addListener(EventObjectListener<EventObject> listener){
        this.publisher.addListener(listener);
    }

    @Override
    public void publish(Object source){
        publish(new DistributedEventObject(source));
    }

    @Override
    public void close() {
        executorService.shutdownNow();
        try {
            this.topicConnection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
